package org.zkpk.spark;

import java.util.List;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;

import scala.Tuple2;

public class SecondarySortDemo {

    public static void main(String[] args) throws Exception {
        // 创建sparkcontext对象，sparkcontext是程序的唯一入口

        SparkConf conf = new SparkConf().setAppName("SecondarySortDemo").setMaster("local");
        JavaSparkContext jsc = new JavaSparkContext(conf);
        // 去掉WARN类的输出信息
        jsc.setLogLevel("WARN");

        // 调用textFile()方法，读取日志文件，这里指定本地磁盘文件

        JavaRDD<String> textfile = jsc.textFile("/home/zkpk/experiment/ex_mqtt.csv");

        // 调用mapTfRDD2Pair方法 将tf映射为键值对
        JavaPairRDD<String, serDemo> pairRdd = mapTfRDD2Pair(textfile);

        // 获取每个时间戳的IAQ、温湿度
        JavaPairRDD<String, serDemo> getlines = aggregateByDeviceID(pairRdd);
        // 聚合，封装的RDD作为key,时间戳作为值
        JavaPairRDD<SecondarySortKey, String> telval = mapRDDKey2SortKey(getlines);

        // 依次按照温湿度、IAQ倒序排序
        JavaPairRDD<SecondarySortKey, String> sortedRdd = telval.sortByKey(false);
        // 根据需要获得输出，显示前200行
        List<Tuple2<SecondarySortKey, String>> getTop = sortedRdd.take(200);
        System.out.println("============ result ============");
        for (Tuple2<SecondarySortKey, String> dt : getTop) {
            System.out.println("TimeStamp: " + dt._2 + ", " + dt._1);
        }

        // 执行结束关闭资源
        jsc.close();
    }

    // mapTfRDD2Pair方法，封装键值对
    private static JavaPairRDD<String, serDemo> mapTfRDD2Pair(JavaRDD<String> accessLogRDD) {
        return accessLogRDD.mapToPair(new PairFunction<String, String, serDemo>() {

            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<String, serDemo> call(String lines) throws Exception {

                if (lines.indexOf(",,") != -1) {
                    lines = lines.substring(0, lines.indexOf(",,") + 1) + "0"
                            + lines.substring(lines.indexOf(",,") + 1);
                }

                // 根据数据格式进行切分
                String[] split = lines.split(",");

                // 获取切分的字段
                String time = split[1];
                Float humidity = Float.valueOf(split[4]);
                Integer IAQ = Integer.valueOf(split[5]);
                Float temperature = Float.valueOf(split[7]);

                // 创建cmbInfo对象，有参构造 ，封装为自定义的可序列化对象
                serDemo dataInfo = new serDemo(humidity, IAQ, temperature);

                return new Tuple2<String, serDemo>(time, dataInfo);
            }

        });
    }

    // 根据时间戳进行聚合，依次按照温度、湿度以及IAQ倒序排序
    private static JavaPairRDD<String, serDemo> aggregateByDeviceID(JavaPairRDD<String, serDemo> accessLogPairRDD) {
        return accessLogPairRDD.reduceByKey(new Function2<serDemo, serDemo, serDemo>() {

            private static final long serialVersionUID = 1L;

            @Override
            public serDemo call(serDemo d1, serDemo d2) throws Exception {
                Integer IAQ = d1.getIAQ() < d2.getIAQ() ? d1.getIAQ() : d2.getIAQ();
                Float humidity = d1.getHumidity() + d2.getHumidity();
                Float temperature = d1.getTemperature() + d2.getTemperature();

                serDemo accessLogInfo = new serDemo();
                accessLogInfo.setIAQ(IAQ);
                accessLogInfo.setHumidity(humidity);
                accessLogInfo.setTemperature(temperature);

                return accessLogInfo;
            }

        });
    }

    // 二次排序，时间戳作为值
    private static JavaPairRDD<SecondarySortKey, String> mapRDDKey2SortKey(
            JavaPairRDD<String, serDemo> aggrAccessLogPairRDD) {
        return aggrAccessLogPairRDD.mapToPair(

                new PairFunction<Tuple2<String, serDemo>, SecondarySortKey, String>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Tuple2<SecondarySortKey, String> call(Tuple2<String, serDemo> tuple) throws Exception {
                        // 获取元组里的数据
                        String telNum = tuple._1;
                        serDemo accessLogInfo = tuple._2;

                        // 封装为二次排序key
                        SecondarySortKey accessLogSortKey = new SecondarySortKey(accessLogInfo.getHumidity(),
                                accessLogInfo.getIAQ(), accessLogInfo.getTemperature());

                        // 返回新的Tuple
                        return new Tuple2<SecondarySortKey, String>(accessLogSortKey, telNum);
                    }
                });
    }
}